package com.yihaodian.search.broker;

import java.util.Collection;
import java.util.Collections;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;

import org.apache.log4j.Logger;

import com.google.common.collect.ArrayListMultimap;
import com.google.common.collect.ListMultimap;
import com.google.common.collect.Lists;
import com.yihaodian.api.index.IndexDoc;
import com.yihaodian.api.index.IndexRequest;
import com.yihaodian.api.search.SearchException;
import com.yihaodian.cluster.ClusterState;
import com.yihaodian.index.meta.IndexMeta;
import com.yihaodian.index.meta.shard.Shard;
import com.yihaodian.index.meta.shard.ShardHandler;
import com.yihaodian.index.meta.shard.impl.ModShardHandler;
import com.yihaodian.search.broker.shard.ShardIndexRequest;

/**
 * routing request to target node. we use hedwig for load-balancer, this class
 * become rather simple.
 */
public class OperationRouting {
	private static Logger log = Logger.getLogger(OperationRouting.class);

	/**
	 * <ClassName, Instance>
	 */
	ConcurrentHashMap<String, ShardHandler> shardHanlerMap = new ConcurrentHashMap<String, ShardHandler>();

	public OperationRouting() {

	}

	/**
	 * 
	 * a search routing contains two steps 1) how many shard of given index 2)
	 * for each shard pick up a node (load balance). We use hedwig, so only
	 * first step is done here.
	 * 
	 * @param request
	 * @return
	 */
	public List<Shard> shardRequest(ClusterState cs, String indexName) {
		List<Shard> targetShards = Lists.newArrayList();

		IndexMeta meta = cs.indexMetas().indexMeta(indexName);
		if (meta == null || meta.getShards() == null
				|| meta.getShards().isEmpty())
			return Collections.EMPTY_LIST;

		Collection<Shard> shards = meta.getShards();
		for (Shard shard : shards) {
			if (shard.isEnable()) {
				targetShards.add(shard);
			} else {
				log.info(shard.getShardId() + " is disable, not search");
			}
		}
		return targetShards;
	}

	public List<ShardIndexRequest> shardRequest(ClusterState cs,
			IndexRequest request) {
		String indexName = request.getIndexName();

		IndexMeta meta = cs.indexMetas().indexMeta(indexName);

		// Get partitioner
		String handleClass = meta.getShardHandlerClass();
		ShardHandler sh = shardHanlerMap.get(handleClass);
		if (sh == null) {
			sh = createShardHandler(handleClass);
			if(sh instanceof ModShardHandler)
			{
				((ModShardHandler)sh).setField(meta.getShardHandlerParam());
			}
			shardHanlerMap.putIfAbsent(handleClass, sh);
		}

		// Split the documents
		ListMultimap<String, IndexDoc> splits = ArrayListMultimap.create();
		Collection<IndexDoc> docs = request.getDocs();
		for (IndexDoc doc : docs) {
			if (!doc.isExternalUpdate()) {
				String id = sh.shard(cs, indexName, doc).getShardId();
				splits.put(id, doc);
			} else {
				// push external data update to every shard.
				for (Shard shard : meta.getShards()) {
					splits.put(shard.getShardId().getShardId(), doc);
				}
			}
		}

		// Construct shard request
		List<ShardIndexRequest> shardRequests = Lists.newArrayList();
		Iterator<String> it = splits.keySet().iterator();
		while (it.hasNext()) {
			String shardId = it.next();
			Collection<IndexDoc> indexDocs = splits.get(shardId);
			ShardIndexRequest shardReq = new ShardIndexRequest(shardId,
					new IndexRequest(indexName, indexDocs));
			shardRequests.add(shardReq);
		}

		return shardRequests;
	}

	ShardHandler createShardHandler(String className) {
		try {
			ShardHandler shardHandler = (ShardHandler) Class.forName(className)
					.newInstance();
			return shardHandler;
		} catch (Exception e) {
			throw new SearchException("can not get ShardHandler" + className, e);
		}
	}
}
